package org.mine.database.migration;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

public class MysqlInsert extends Thread {
	final static String driver = "com.mysql.jdbc.Driver";
	final static String url = "jdbc:mysql://192.168.68.140:3306/mapi?zeroDateTimeBehavior=convertToNull&amp;autoReconnect=true&amp;failOverReadOnly=false&amp;autoReconnectForPools=true";
	final static String user = "xieweili_w";
	final static String pwd = "ppjyG9xfecQpm_j";
	
    static PreparedStatement statement = null;
    
    private AtomicLong total;
    
    Connection connection = null;
    
    static{
    	 try {
			Class.forName(driver);
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
    }
	
	BlockingQueue<DataModel> queue;
	public MysqlInsert(BlockingQueue<DataModel> q, AtomicLong total) throws Exception {
		connection = DriverManager.getConnection(url, user, pwd);
		this.queue = q;
		this.total = total;
	}

	public void run() {
		try {
			while (true) {
				long t1 = System.currentTimeMillis();
				consume(queue.take());
				long t2 = System.currentTimeMillis();
				System.out.println(Thread.currentThread().getName() + "执行耗时" + (t2 - t1) + ", 当前等待任务数:" + queue.size());
			}
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public void consume(DataModel dataModel) throws Exception {
		insertAll(dataModel.getSql(), dataModel.getData());
	}
	
	/**
     * 执行数据库插入操作
     *
     * @param datas     插入数据表中key为列名和value为列对应的值的Map对象的List集合
     * @param tableName 要插入的数据库的表名
     * @return 影响的行数
     * @throws SQLException SQL异常
     */
    private void insertAll(String sql, List<List<Object>> datas) throws SQLException {
        /**影响的行数**/
        int affectRowCount = -1;
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        List<Object> pkids = new ArrayList<>();
        try {
            /**设置不自动提交，以便于在出现异常的时候数据库回滚**/
            connection.setAutoCommit(false);
            datas.forEach(x->{
            	pkids.add(x.get(0));
            	for (int j = 0; j < x.size(); j++) {
					try {
						preparedStatement.setObject(j + 1, x.get(j));
					} catch (SQLException e) {
						e.printStackTrace();
					}
                }
            	try {
					preparedStatement.addBatch();
				} catch (SQLException e) {
					e.printStackTrace();
				}
            });
            
            int[] arr = preparedStatement.executeBatch();
            connection.commit();
            affectRowCount = arr.length;
            long remaine = total.addAndGet(affectRowCount * -1);
            System.out.println(Thread.currentThread().getName() + "成功了插入了" + affectRowCount + "行, 剩余" + remaine);
        } catch (Exception e) {
            if (connection != null) {
                connection.rollback();
            }
            for(Object iObject : pkids){
            	System.out.print(iObject+",");
            }
            e.printStackTrace();
            throw e;
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
    }
}
